package edu.hawaii.ics.pjdb.worker.command_runners;

import java.sql.Connection;
import java.sql.SQLException;
import java.util.List;

import javax.sql.rowset.CachedRowSet;

import org.apache.thrift.TException;
import org.apache.thrift.server.TServer;
import org.apache.thrift.transport.TTransportException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.sun.rowset.CachedRowSetImpl;

import edu.hawaii.ics.pjdb.result_transport.TupleListener;
import edu.hawaii.ics.pjdb.result_transport.TupleServiceHelper;
import edu.hawaii.ics.pjdb.types.DbSyncRowSetCommand;
import edu.hawaii.ics.pjdb.types.Tuple;
import edu.hawaii.ics.pjdb.types.WorkerCommand;

public class DbSyncRowSetRunner extends CommandRunner {
    private static final Logger logger = LoggerFactory
            .getLogger(DbSyncRowSetRunner.class);

    /**
     * Handles input tuples and executes the queries and discards the query
     * results (results should not be expected from DbSyncRowSet commands).
     * 
     */
    protected static class SyncTupleListener extends TupleListener {
        private static final Logger logger = LoggerFactory
                .getLogger(SyncTupleListener.class);

        private final DbSyncRowSetCommand cmd;
        private final Connection localDb;

        public SyncTupleListener(WorkerCommand workerCmd, Connection localDb)
                throws SQLException, TTransportException, InterruptedException {
            this.localDb = localDb;
            this.cmd = workerCmd.getCommand().getDbSyncRowSetCmd();
            logger.info("Stuff is all set up. Listening for tuples.");
        }

        @Override
        public void receiveTuples(List<Tuple> tuples) {
            logger.debug("#receiveTuples() called.");
            try {

                // Set-up RowSet.
                CachedRowSet crset = new CachedRowSetImpl();
                crset.setConcurrency(CachedRowSet.CONCUR_UPDATABLE);
                crset.setCommand("SELECT * FROM " + cmd.getTableName());
                crset.execute(localDb);
                crset.moveToInsertRow();
                for (Tuple tuple : tuples) {
                    logger.info("Received tuple: {}", tuple.getEntries());

                    // Set row parameters
                    int parameterIndex = 1;
                    for (String fieldValue : tuple.getEntries()) {
                        crset.updateObject(parameterIndex, fieldValue);
                        parameterIndex++;
                    }

                    // Insert row to the RowSet
                    crset.insertRow();
                }

                // Make changes to database.
                crset.moveToCurrentRow();
                crset.acceptChanges(localDb);
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }

        @Override
        public void finished() {
            logger.info("#finish() called.");
        }

    }

    public DbSyncRowSetRunner(Connection localDb) {
        super(localDb);
    }

    @Override
    public void execute(WorkerCommand cmd) throws TException, SQLException {
        logger.info(
                "Received DbSyncRowSet command. Setting up input listener on port {}",
                cmd.getCommand().getDbSyncRowSetCmd().getInputAddress()
                        .getPort());

        try {
            SyncTupleListener syncWorker = new SyncTupleListener(cmd, localDb);
            TServer inputServer = TupleServiceHelper.createNonThreadedServer(
                    cmd.getCommand().getDbSyncRowSetCmd().getInputAddress()
                            .getPort(), syncWorker);
            inputServer.serve();
        } catch (InterruptedException e) {
            throw new TException(e);
        }
    }
}
